package com.ib.algokit.marketscan;

import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.activequant.broker.IBroker;
import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.account.BrokerAccount;
import org.activequant.core.domainmodel.account.Position;
import org.activequant.core.domainmodel.marketscanner.InstrumentSpecificationDetails;
import org.activequant.core.domainmodel.marketscanner.MarketScanChange;
import org.activequant.core.types.Currency;
import org.activequant.core.types.SecurityType;
import org.activequant.core.types.TimeStamp;
import org.activequant.math.algorithms.Interval;
import org.activequant.tradesystem.types.Mode;
import org.apache.log4j.Logger;

import com.ib.algokit.AnalyzerUtils;
import com.ib.algokit.ForexAnalyzerUtils;
import com.ib.algokit.InstrumentAnalyzer;
import com.ib.algokit.RunMe;
import com.ib.algokit.TraderProxy;
import com.ib.algokit.ib.IBPortfolioManagingBroker;
import com.ib.algokit.ib.IBPortfolioManagingBroker.IBPosition;
/**
 * ThreadPerInstrumentBaseStrategy implements IEventListener&lt;MarketScanChange&gt;.<br/>
 * This is the listener for MarketScanChanges that starts a separate Thread to analyze an Instrument that seems interesting.<br/>
 * Holds the associated ema(EMA) variable. It also defines a nested InstrumentAnalyzingThread class.
 * @author dmisev
 *
 */
public class ThreadPerInstrumentBaseStrategy {
	/**
	 * protected final Logger log = Logger.getLogger(getClass());
	 */
	protected final Logger log = Logger.getLogger(getClass());
	
	protected int MAX_ALLOWED_INSTRUMENTS=100;//for testing
	/**
	 * public static final double PROFIT_GOAL_RATIO=0.002d;
	 */
	public static final double PROFIT_GOAL_RATIO=0.002d;
	/**
	 * protected final EMA ema;
	 */
	protected final TraderProxy ema;
	/**
	 * protected final Map&lt;InstrumentSpecification, InstrumentAnalyzingThread&gt; analyzerThreads = new ConcurrentHashMap&lt;InstrumentSpecification, InstrumentAnalyzingThread&gt;();
	 */
//	protected final Map<InstrumentSpecification, InstrumentAnalyzingThread> analyzerThreads = new ConcurrentHashMap<InstrumentSpecification, InstrumentAnalyzingThread>();
	/**
	 * Constructs a ThreadPerInstrumentEventHandlingStrategy(implements IEventListener&lt;MarketScanChange&gt;) using the given ema(EMA) to set its
	 * associated ema(EMA).<br/>
	 * It also iterates the existing Positions and starts an InstrumentAnalyzingThread for those whose quantity is not 0
	 * @param ema
	 */
	public ThreadPerInstrumentBaseStrategy(TraderProxy ema){
		this.ema=ema;
//		InstrumentSpecification spec=new InstrumentSpecification(new Symbol("AAPL"), "SMART", "IB", Currency.USD, SecurityType.OPTION);
//		analyzeInstrument(spec);
		
//		analyzePositions();		
	}
	/**
	 * This iterates the existing Positions and starts an InstrumentAnalyzingThread for those whose quantity is not 0.
	 * Gets called only once at the start.
	 */
	public void analyzePositions() {
		IBroker broker = this.ema.getBroker();
		if (broker instanceof IBPortfolioManagingBroker) {
			IBPortfolioManagingBroker IBAccountManagingBroker = (IBPortfolioManagingBroker) broker;
			BrokerAccount brokerAccount = IBAccountManagingBroker.getBrokerAccount();
			Iterator<Position> iterator = brokerAccount.getPortfolio().getPositionIterator();
			IBPosition pos = null;
			Position position = null;
			while (iterator.hasNext()) {
				position = iterator.next();
				if (position instanceof IBPosition) {
					pos = (IBPosition) position;
					if(pos.getQuantity()==0){
						continue;
					}
					double unrealizedPNL=0;
					try{
						unrealizedPNL=Double.parseDouble(pos.getAdditionalInformation().get(IBPortfolioManagingBroker.UNREALIZED_PNL));
					}catch(Exception e){
					}
					//if it is losing money, don't even look at it
					if(unrealizedPNL<0){
						continue;
					}
					InstrumentSpecification instrumentSpecification = pos.getInstrumentSpecification();
					if (ema.getAnalyzerThreads().size() >= MAX_ALLOWED_INSTRUMENTS)// for testing only
						return;
					InstrumentAnalyzingThread instrumentAnalyzingThread = ema.getAnalyzerThreads().get(instrumentSpecification);
					if (instrumentAnalyzingThread == null) {
						//instrumentSpecification=ema.getInstrumentSpecificationDao().update(instrumentSpecification);	
						instrumentAnalyzingThread = new InstrumentAnalyzingThread(instrumentSpecification, true);
						ema.getAnalyzerThreads().put(instrumentSpecification, instrumentAnalyzingThread);
						startAnalyzerThread(instrumentAnalyzingThread);
					}
				}
			}
		}
	}
	/**
	 * this gets called (but not at the first time positions are loaded) by the IBPortfolioManagingBroker for positions whose profitRation has been reached.
	 * It will only do something if the position isn't already being analyzed.
	 * @param pos
	 */
	public void updateAndAnalyzePosition(IBPosition pos) {		
		InstrumentSpecification instrumentSpecification = pos.getInstrumentSpecification();
		InstrumentAnalyzingThread instrumentAnalyzingThread = ema.getAnalyzerThreads().get(instrumentSpecification);
		String symbol=instrumentSpecification.getSymbol().toString();
		SecurityType securityType=instrumentSpecification.getSecurityType();
		Currency currency = instrumentSpecification.getCurrency();
		if (instrumentAnalyzingThread == null) {//this should happen if i am not already analyzing the position
			StringBuffer msg = new StringBuffer("initializing analyzer for "+symbol);
			if(SecurityType.CASH==securityType){
				msg.append("/"+currency);
			}
			log.info(msg.toString());
			instrumentAnalyzingThread = new InstrumentAnalyzingThread(instrumentSpecification, SecurityType.CASH!=securityType);
			ema.getAnalyzerThreads().put(instrumentSpecification, instrumentAnalyzingThread);
			startAnalyzerThread(instrumentAnalyzingThread);
		}else{
			StringBuffer msg = new StringBuffer("analyzer for "+symbol);
			if(SecurityType.CASH==securityType){
				msg.append("/"+currency);
			}
			msg.append(" already initialized");
			log.info(msg.toString());			
		}
		if(SecurityType.CASH==securityType){
			InstrumentAnalyzer instrumentAnalyzer=instrumentAnalyzingThread.instrumentAnalyzer;
			ForexAnalyzerUtils.closeForexPositionIfNotWorkingAnything(pos,instrumentAnalyzer);
		}
	}
	/**
	 * Analyze the given instrumentSpecification(InstrumentSpecification)
	 * @param instrumentSpecification
	 */
	public void analyzeInstrument(InstrumentSpecification instrumentSpecification) {
		InstrumentAnalyzingThread instrumentAnalyzingThread = ema.getAnalyzerThreads().get(instrumentSpecification);
		if (instrumentAnalyzingThread == null) {
			// instrumentSpecification=ema.getInstrumentSpecificationDao().update(instrumentSpecification);
			instrumentAnalyzingThread = new InstrumentAnalyzingThread(instrumentSpecification);
			ema.getAnalyzerThreads().put(instrumentSpecification, instrumentAnalyzingThread);
			startAnalyzerThread(instrumentAnalyzingThread);
		}
	}
	/**
	 * Starts the given instrumentAnalyzingThread(InstrumentAnalyzingThread)
	 * @param instrumentAnalyzingThread
	 */
	protected void startAnalyzerThread(InstrumentAnalyzingThread instrumentAnalyzingThread) {
		instrumentAnalyzingThread.start();
	}
	/**
	 * If the instrument data loading isn't finished yet, waits until it is and then cancels all live data feeds and removes the InstrumentAnalyzingThread.
	 * If the given completelyRemoveInstrumentAnalyzer(boolean) is true it also removes the InstrumentAnaluzer with all the data in it<br/>
	 * You shouldn't try to remove the InstrumentAnyzer and leave the Thread running, this will kick InvalidArgumentException.<br/>
	 * You may however stop the Thread and optionally completely remove the InstrumentAnalyzer from memory, to save some time if we need to kick this again
	 * @param instrumentSpecification
	 * @param stopAnalyzingAndCancelAllFeeds this will also in effect stop placing orders trading because I place them on events, except for updatePortfolio calls
	 * @param completelyRemoveInstrumentAnalyzer
	 * @return returns whether an InstrumentAnalyzer was actually removed
	 */
	public boolean removeInstrumentAnalyzer(InstrumentSpecification instrumentSpecification, boolean stopAnalyzingAndCancelAllFeeds, boolean completelyRemoveInstrumentAnalyzer){
		if(!stopAnalyzingAndCancelAllFeeds&&completelyRemoveInstrumentAnalyzer){
			throw new IllegalArgumentException("You can't remove the InstrumentAnalyzer without stopping the Thread first");
		}
		boolean removedThreadAndCancelledAllFeeds=false;
		InstrumentAnalyzingThread instrumentAnalyzingThread = ema.getAnalyzerThreads().get(instrumentSpecification);
		if (stopAnalyzingAndCancelAllFeeds && instrumentAnalyzingThread != null) {
			instrumentAnalyzingThread.cancelLiveFeeds();
			ema.getAnalyzerThreads().remove(instrumentSpecification);
			removedThreadAndCancelledAllFeeds=true;			
		}
		boolean removedInstrumentAnalyzer=false;
		if(completelyRemoveInstrumentAnalyzer){
			removedInstrumentAnalyzer=(InstrumentAnalyzer.getInstrumentAnalyzers().remove(instrumentSpecification)!=null);
		}
		return removedThreadAndCancelledAllFeeds;
	}	
	/**
	 * InstrumentAnalyzingThread extends Thread. A nested class defined within the ThreadPerInstrumentEventHandlingStrategy class.<br/>
	 * Holds the following associated variables:
	 * <ul>
	 * <li>instrumentAnalyzerInitialized(AtomicBoolean)</li>
	 * <li>instrumentAnalyzer(InstrumentAnalyzer)</li>
	 * <li>marketScanChanges(BlockingQueue&lt;MarketScanChange&gt;)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	public class InstrumentAnalyzingThread extends Thread {
		/**
		 * private AtomicBoolean instrumentAnalyzerInitialized=new AtomicBoolean();
		 */
		private AtomicBoolean instrumentAnalyzerInitialized=new AtomicBoolean();
		/**
		 * final variable, holds the associated instrumentAnalyzer(InstrumentAnalyzer)<br/>
		 * , initialized by the constructor to an InstrumentAnalyzer for a given instrumentSpecification(InstrumentSpecification)
		 */
		private final InstrumentAnalyzer instrumentAnalyzer;
		/**
		 * final variable, holds the associated marketScanChanges(BlockingQueue&lt;MarketScanChange&gt;)<br/>
		 * , initialized to a new LinkedBlockingQueue&lt;MarketScanChange&gt;
		 */
		private final BlockingQueue<MarketScanChange> marketScanChanges = new LinkedBlockingQueue<MarketScanChange>();
		/**
		 * Constructs a InstrumentAnalyzingThread(Thread subclass) using the given instrumentSpecification(InstrumentSpecification) and a null startingPosition(IBPosition) to create and set set its associated instrumentAnalyzer(InstrumentAnalyzer).
		 * It also sets the name of the Thread to "NEO Thread for ${instrumentSpecification.getSymbol()}"
		 * @param instrumentSpecification
		 */
		public InstrumentAnalyzingThread(InstrumentSpecification instrumentSpecification) {
			this(instrumentSpecification, false);
		}
		/**
		 * Constructs a InstrumentAnalyzingThread(Thread subclass) using the given instrumentSpecification(InstrumentSpecification) and startingPosition(IBPosition) to create and set set its associated instrumentAnalyzer(InstrumentAnalyzer).
		 * It also sets the name of the Thread to "NEO Thread for ${instrumentSpecification.getSymbol()}"
		 * @param instrumentSpecification
		 * @param startingPosition
		 */
		public InstrumentAnalyzingThread(InstrumentSpecification instrumentSpecification, boolean closePositionAndStopAnalyzing) {	
			super();
			if(instrumentSpecification.getSecurityType()==SecurityType.CASH){
				super.setName("NEO Thread for " + instrumentSpecification.getSymbol()+"/"+instrumentSpecification.getCurrency());
			}else{
				super.setName("NEO Thread for " + instrumentSpecification.getSymbol());
			}
			this.instrumentAnalyzer = new InstrumentAnalyzer(ema, instrumentSpecification, closePositionAndStopAnalyzing);
		}
		/**
		 * This constructor is for when I already have InstrumentSpecificationDetails from the marketScanner (also saved in the database so we can skip that step in the thread)
		 * @param instrumentSpecificationDetails
		 * @param closePositionAndStopAnalyzing
		 */
		public InstrumentAnalyzingThread(InstrumentSpecificationDetails instrumentSpecificationDetails, boolean closePositionAndStopAnalyzing) {
			super();
			InstrumentSpecification instrumentSpecification = instrumentSpecificationDetails.getInstrumentSpecification();
			if(instrumentSpecification.getSecurityType()==SecurityType.CASH){
				super.setName("NEO Thread for " + instrumentSpecification.getSymbol()+"/"+instrumentSpecification.getCurrency());
			}else{
				super.setName("NEO Thread for " + instrumentSpecification.getSymbol());
			}
			this.instrumentAnalyzer = new InstrumentAnalyzer(ema,instrumentSpecification, instrumentSpecificationDetails, closePositionAndStopAnalyzing);
		}
		/**
		 * Adds/puts the given marketScanChange(MarketScanChange) into the associated marketScanChanges(BlockingQueue&lt;MarketScanChange&gt;) 
		 * (this is a blocking operation, if there is no room it waits until there is a free slot).<br/>
		 * If an e(InterruptedException) occurs, logs the error into the log(Log) of the associated instrumentAnalyzer(InstrumentAnalyzer)
		 * @param message
		 */
		public void enqueue(MarketScanChange marketScanChange) {
			try {
				marketScanChanges.put(marketScanChange);
			} catch (InterruptedException e) {
				instrumentAnalyzer.getLog().error(e.getMessage());
			}
		}

		public void run() {
			if(this.instrumentAnalyzer.getInstrumentSpecification().getSecurityType()==SecurityType.CASH){
//				if(!RunMe.isRealRun){//i don't want replay of a day worth of data
//					return;
//				}
				
				if(!this.instrumentAnalyzer.getState().isLiveQuoteFeedActivated())
				{
					this.instrumentAnalyzer.subscribeToLiveQuoteFeed();	
				}
				return;
			}
			this.instrumentAnalyzer.requestInstrumentSpecificationDetails();
			if(this.instrumentAnalyzer.gotMultipleInstrumentSpecsDetails()){
				//do something to load a separate instrument analyzer for each				
				return;
			}
//			this.instrumentAnalyzer.getT_MINUS_5D_HistoryMarketScans();
//			if(false&&!RunMe.isRealRun){
//				this.instrumentAnalyzer.initAllMarketScanHistory();
//			}
			this.instrumentAnalyzer.getT_MINUS_1Y_HistoryCandles();
			if(!this.instrumentAnalyzer.getState().isCandleT_MINUS_1YHistoryRequestReceivedOK()){
				return;
			}
			
			this.instrumentAnalyzer.getT_MINUS_5D_HistoryCandles();
			if(!this.instrumentAnalyzer.getState().isCandleT_MINUS_5DHistoryRequestReceivedOK()){
				return;
			}
			
			this.instrumentAnalyzer.getT_MINUS_5D_HistoryTickerNews();
//			if(!this.instrumentAnalyzer.getState().isTickerNewsT_MINUS_5DHistoryRequestReceivedOK()){
//				return;
//			}
			//custom historical data processing code goes in this method			
			this.instrumentAnalyzer.historicalDataProcessing();
			//custom code end
			
			if(RunMe.mode==Mode.BACKTEST){//i don't want replay of a day worth of data
				return;
			}
			
			
			if(!this.instrumentAnalyzer.getState().isLiveQuoteFeedActivated())
			{
				this.instrumentAnalyzer.subscribeToLiveQuoteFeed();	
			}			
			if(!this.instrumentAnalyzer.getState().isLiveCandleFeedActivated())
			{
				this.instrumentAnalyzer.subscribeToLive5SecondCandleFeed();
			}			
			if(!this.instrumentAnalyzer.getState().isLiveTradeIndicationFeedActivated())
			{
				this.instrumentAnalyzer.subscribeToLiveTradeIndicationFeed();
			}			
			if(RunMe.isMarketDepthEnabled&&!this.instrumentAnalyzer.getState().isLiveMarketDepthFeedActivated())
			{
				this.instrumentAnalyzer.subscribeToLiveMarketDepthFeed();	
			}
			
			if(!this.instrumentAnalyzer.getState().isLiveTickerNewsFeedActivated())
			{
				this.instrumentAnalyzer.subscribeToLiveTickerNewsFeed();	
			}
			
			this.instrumentAnalyzer.closePositionIfProfitGoalReached();
			
			
			
			//just info
			Interval<TimeStamp> T_MINUS_5D_INTERVAL = this.instrumentAnalyzer.getExchangeTradingRange_T_MINUS_5(this.instrumentAnalyzer.getExchangeCalendar());
			this.instrumentAnalyzer.getLog().info("T_MINUS_5D_INTERVAL: "+InstrumentAnalyzer.fmt.format(T_MINUS_5D_INTERVAL.from)+" "+InstrumentAnalyzer.fmt.format(T_MINUS_5D_INTERVAL.to));
			
			//just info
			Interval<TimeStamp> T_MINUS_1Y_INTERVAL = this.instrumentAnalyzer.getExchangeTradingRange_T_MINUS_1Y(this.instrumentAnalyzer.getExchangeCalendar());
			this.instrumentAnalyzer.getLog().info("T_MINUS_1Y_INTERVAL: "+InstrumentAnalyzer.fmt.format(T_MINUS_1Y_INTERVAL.from)+" "+InstrumentAnalyzer.fmt.format(T_MINUS_1Y_INTERVAL.to));
//			if(this.instrumentAnalyzer.getAnalyzerPlugin() instanceof AnalyzerUtils){
//			AnalyzerUtils analyzerUtils=(AnalyzerUtils)this.instrumentAnalyzer.getAnalyzerPlugin();
//			analyzerUtils.getTMINUS5DHighAndLowCandles();
////			
//			analyzerUtils.getTMINUS1YHighAndLowCandles();
//			}
			
//			this.instrumentAnalyzer.getAnalyzerPlugin().initializeTMINUS5DCandles(Integer.MAX_VALUE);
			

//			
//			this.instrumentAnalyzer.getAnalyzerPlugin().findExtremes2(1);
//			
//			this.instrumentAnalyzer.getAnalyzerPlugin().drawTMINUS5DCandles();
			
//			this.instrumentAnalyzer.getAnalyzerPlugin().calculateChange();
//			
//			this.instrumentAnalyzer.getAnalyzerPlugin().initializeBollingerBands(20, 2,false);
//			
//			this.instrumentAnalyzer.getAnalyzerPlugin().initializeBollingerBandsFromDailyCandles(20, 2, false);
//			
//			
//			
//			
//			
//			this.instrumentAnalyzer.getAnalyzerPlugin().initializeCandleDirection();
//			this.instrumentAnalyzer.getAnalyzerPlugin().updateCandleVolumeDirection();
//			this.instrumentAnalyzer.getAnalyzerPlugin().updateCandleTradeCountDirection();
//			
//			this.instrumentAnalyzer.getAnalyzerPlugin().getRelativeStrengthIndex();

			if(instrumentAnalyzerInitialized.getAndSet(true)){
				instrumentAnalyzer.getLog().warn("this instrument was already initialized before");
			}
			synchronized (instrumentAnalyzerInitialized) {
				instrumentAnalyzerInitialized.notifyAll();
			}
//			if(this.instrumentAnalyzer.getClosePositionAndDontTrade()){
//				removeInstrumentAnalyzer(this.instrumentAnalyzer.getInstrumentSpecification(), true, true);
//			}
		}
		
		public void cancelLiveFeeds(){			
			while(!instrumentAnalyzerInitialized.getAndSet(false)){
				synchronized (instrumentAnalyzerInitialized) {
					try {
						instrumentAnalyzerInitialized.wait();
					} catch (InterruptedException ex) {
						instrumentAnalyzer.getLog().info("I woke up and instrumentAnalyzerInitialized=" + instrumentAnalyzerInitialized.get());
					}
				}
			}
			
			if(this.instrumentAnalyzer.getState().isLiveQuoteFeedActivated())
			{
				this.instrumentAnalyzer.cancelLiveQuoteFeed();	
			}			
			if(this.instrumentAnalyzer.getState().isLiveCandleFeedActivated())
			{
				this.instrumentAnalyzer.cancelLive5SecondCandleFeed();
			}			
			if(this.instrumentAnalyzer.getState().isLiveTradeIndicationFeedActivated())
			{
				this.instrumentAnalyzer.cancelLiveTradeIndicationFeed();
			}			
			if(this.instrumentAnalyzer.getState().isLiveMarketDepthFeedActivated())
			{
				this.instrumentAnalyzer.cancelLiveMarketDepthFeed();	
			}
			if(this.instrumentAnalyzer.getState().isLiveTickerNewsFeedActivated())
			{
				this.instrumentAnalyzer.cancelLiveTickerNewsFeed();	
			}
		}
	}
}